package cn.com.zetatech.maple.core;

import cn.com.zetatech.maple.core.resource.FtpLoader;
import cn.com.zetatech.maple.core.resource.RuleLoader;
import cn.com.zetatech.maple.core.task.ParseTask;
import cn.com.zetatech.maple.core.utils.ConfigurationManager;
import cn.com.zetatech.maple.entity.FtpDFile;
import cn.com.zetatech.maple.entity.Rule;
import cn.com.zetatech.maple.rule.RuleParser;
import com.google.common.base.Joiner;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author cheng.cao@zetatech.com.cn
 */
@Slf4j
@SuppressWarnings("all")
public class Bootstrap {
    private static final DateTimeFormatter fmt = DateTimeFormatter.ofPattern("yyyyMMdd");

    public static void main(String[] args) throws Exception {
        Stopwatch stopwatch = Stopwatch.createStarted();
        String MAPLE_HOME = System.getProperty("MAPLE_HOME");
        if (StringUtils.isBlank(MAPLE_HOME)) {
            log.error("环境变量{}未设置", "MAPLE_HOME");
            System.exit(-1);
        }
        FutureTask<Integer> ruleLoaderFuture = new FutureTask<>(() -> RuleLoader.load());
        FutureTask<List<FtpDFile>> dFileLoaderFuture = new FutureTask<>(() -> FtpLoader.load());

        new Thread(ruleLoaderFuture, "thread-load-rules").start();
        new Thread(dFileLoaderFuture, "thread-download-file").start();

        if (ruleLoaderFuture.get() == 0) {
            log.error("由于未成功加载到任何规则文件，故直接退出系统");
            return;
        }
        List<FtpDFile> ftpDFiles = dFileLoaderFuture.get();
        if (ftpDFiles.isEmpty()) {
            log.error("由于未成功下载到任何日志文件，故直接退出系统");
            return;
        }
        String dirPattern = ConfigurationManager.me().getString("ftp.dir-pattern");
        Pattern p = Pattern.compile(dirPattern);

        String rootDir = ConfigurationManager.me().getString("ftp.log-base-dir");
        Map<String, FtpDFile> mergedFiles = Maps.newHashMap();
        for (FtpDFile merge : ftpDFiles) {
            if (!merge.isSuccess()) {
                continue;
            }

            // 开关： 是否需要本地将同类型的文件做文件合并，如果不需要，则key为: 本地文件路径(否则key:${设备类型}_${日志类型})
            {
                Matcher m = p.matcher(merge.getRemote());
                List<String> keys = new ArrayList<>();
                while (m.find() && keys.size() < 2) {
                    keys.add(m.group().replaceAll("\\d+", ""));
                }
                String key = Joiner.on("_").join(keys);
                Rule rule = RuleParser.getFields(key);

                if (rule == null) {
                    log.error("文件{}未找到对应的解析规则，忽略!", merge.getRemote());
                    continue;
                }
                // 开关：是否需要本地将同类型的文件做文件合并，如果不需要，则key为: 本地文件路径
                if (!rule.isEnableSameTypeLocalMerge()) { // 不需要本地合并
                    mergedFiles.put(merge.getLocal(), merge);
                    continue;
                }
            }


            String dir = merge.getRemote().replaceFirst(rootDir, "");
            String key = dir.substring(0, dir.lastIndexOf("/")).replaceAll("\\d+", "").replace("/", "_");
            FtpDFile previousFile = mergedFiles.get(key);

            if (previousFile != null) {
                log.info("本地文件合并 {} --> {}", merge.getRemote(), previousFile.getRemote());
                File mergeTo = new File(previousFile.getLocal());
                // 文件合并
                File f = new File(merge.getLocal());
                List<String> lines = FileUtils.readLines(f, StandardCharsets.UTF_8);
                FileUtils.writeLines(mergeTo, lines, true);
                f.delete();
            } else {
                // 这样处理是由于怕文件，最后不是一个换行，导致下一个文件在合并的时候，2行写成了一行，所以先读取出来再写回去
                File mergeTo = new File(merge.getLocal());
                List<String> lines = FileUtils.readLines(mergeTo, StandardCharsets.UTF_8);
                FileUtils.writeLines(mergeTo, lines, false);
                mergedFiles.put(key, merge);
            }
        }

        int parseThreadCoreNum = (int) (Math.ceil(Runtime.getRuntime().availableProcessors() * 0.7));
        if (ftpDFiles.size() < parseThreadCoreNum) {
            parseThreadCoreNum = ftpDFiles.size();
        }
        ExecutorService processPool = Executors.newFixedThreadPool(parseThreadCoreNum);


        CompletionService<List<Map<String, String>>> exec = new ExecutorCompletionService<>(processPool);
        for (FtpDFile ftpDFile : mergedFiles.values()) {
            ParseTask parseTask = new ParseTask(p, ftpDFile);
            exec.submit(parseTask);
        }

        for (int i = 0; i < mergedFiles.values().size(); i++) {
            try {
                List<Map<String, String>> result = exec.take().get();
            } catch (Exception e) {
                log.error("get data error", e);
            }
        }
        processPool.shutdown();
        System.out.println("任务完成，耗时：" + stopwatch.elapsed(TimeUnit.MILLISECONDS) + "ms");
    }
}
